package com.ljh.seckill.mq;

import com.ljh.seckill.config.SeckillConfig;
import com.ljh.seckill.entity.Order;
import com.ljh.seckill.service.OrderService;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @Author grung
 * @Date 2021/2/23 17:51
 * @Version 1.0
 */

@Component
public class ConsumerListener implements MessageListenerConcurrently {

    private Logger logger= LoggerFactory.getLogger(ConsumerListener.class);


    @Autowired
    StringRedisTemplate redisTemplate;

    @Autowired
    Producer producer;

    @Autowired
    OrderService orderService;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

        //获取消息，查看消息中的订单状态
        MessageExt messageExt=list.get(0);
        String tag = messageExt.getTags();
        String body = new String(messageExt.getBody());
        logger.info("接到消息：tag:{}，body:{}",tag,body);
        if (StringUtils.isBlank(body)){
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

        switch (tag){
            case SeckillConfig.ROCKETMQ_TAG_ORDER_CREATE: {
                String orderId = body.split(",")[0];
                String resourceId = body.split(",")[1];

                // 创建订单，扣减库存
                orderService.createOrder(orderId, resourceId);
                //发送一条延迟消息，1分钟超时取消订单并恢复库存
                Boolean result = producer.sendDelayMsg(SeckillConfig.ROCKETMQ_TOPIC_ORDER,
                        SeckillConfig.ROCKETMQ_TAG_ORDER_CANCEL,
                        orderId, DelayTimeLevel.MINUTES_ONE);

                if (!result) {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }

                break;
            }

            //取消订单
            case SeckillConfig.ROCKETMQ_TAG_ORDER_CANCEL: {
                String orderId = body;
                Boolean result = orderService.cancelOrder(orderId);
                if (result) {
                    //恢复缓存的库存
                    Order order = orderService.findOrderById(orderId);
                    redisTemplate.opsForValue().increment(SeckillConfig.RESOURCE_STOCK + order.getResourceId());
                }
                break;
            }

            default:
                logger.info("消息消费失败，tag not found，msgId-{}", messageExt.getMsgId());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

        logger.info("消息消费成功");
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
